﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

using STSdb.Data;
using STSdb.MTL.Utils.Async;

namespace STSdb.MTL
{
    public class StsTablesCompactor
    {
        /// <summary>
        /// Minimal quantity of changed records in the table.
        /// For starting Compaction procedure.
        /// Note, the compaction will be made only in case if conditions are corresponding to compaction strategy.
        /// </summary>
        public static int _MIN_QUANTITY_FOR_COMPACTION_START = 100;

        /// <summary>
        /// Sets up difference between table compaction. 
        /// Compactor will compare difference in minutes between previous compaction and current time (in UTC).
        /// If difference will be bigger then this value, compaction will start. 
        /// </summary>
        public static int COMPACTION_DIFF_BETWEEN_COMPACTS_IN_MINUTES = 60*4;

        /// <summary>
        /// If you want to receive tables compacting result, fill this with delegate.
        /// </summary>
        public static Action<TableCompactingResult> TablesCompactingResultCollector = null;

        /// <summary>
        /// Indicates that User tables compactor is activated
        /// </summary>
        public static bool COMPACTOR_IS_ACTIVATED = true;

        /// <summary>
        /// Public procedure, who collets TablesCompacting Result
        /// </summary>
        /// <param name="tcs"></param>
        public void TablesCompactingResultHasArrived(TableCompactingResult tcs)
        {
            if (StsTablesCompactor.TablesCompactingResultCollector != null)
            {
                STSdb.MTL.Utils.Async.AsyncOperations.DoAsync(TablesCompactingResultCollector, tcs);
                //TablesCompactingResultCollector.DoAsync(tcs);
            }
        }

        TransactionsCoordinator _transactionsCoordinator = null;

        public StsTablesCompactor(TransactionsCoordinator transactionsCoordinator)
        {
            this._transactionsCoordinator = transactionsCoordinator;
        }

        /// <summary>
        /// Holds Reference to the RawFile where compacting routines are collected.
        /// </summary>
        public Files.RawFile CompacterRawFileReference = null;

        //For writing and reading from the raw file
        ReaderWriterLockSlim _sync = new ReaderWriterLockSlim();

        /// <summary>
        /// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!   MUST BE FILLED
        /// Here we must retriev bytes from the Raw File always multiple of 64,128,256 or whatever
        /// </summary>
        /// <param name="originalUserTableName"></param>
        /// <param name="ptr"></param>
        /// <returns></returns>
        public string GetUsersTableName(string originalUserTableName, ulong pos)
        {
            try
            {
                TableCompactingState tcs = ReadCompactingState(pos);

                if (tcs.ActiveTable == 1)
                {
                    return "@sys_etn_" + originalUserTableName + "/1";
                }
                else
                {
                    return originalUserTableName;

                }
            }
            catch (Exception ex)
            {
                throw ex;
            }

        }

        /// <summary>
        /// When table migrates to the schema Compacter reserves place for it in his rawfile and 
        /// returns position inside of the file as a releative information
        /// </summary>
        /// <returns></returns>
        public ulong NewTableCreatedInSchema()
        {
            try
            {
                ulong pos = CompacterRawFileReference.Length;

                if (pos == 0)
                    pos = (ulong)TablesCompactorHeader._CompactorHeaderSize;

                this.WriteCompactingState(pos, new TableCompactingState());

                return pos;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        /// <summary>
        /// Sets up compacter record on first file
        /// </summary>
        /// <param name="pos"></param>
        public void DeleteTable(ulong pos)
        {
            WriteCompactingState(pos, new TableCompactingState()
            {
                ActiveTable = 0
            });
        }



        /// <summary>
        /// Internal for reading compaction state
        /// </summary>
        /// <param name="pos"></param>
        /// <returns></returns>
        private TableCompactingState ReadCompactingState(ulong pos)
        {
            byte[] rb = new byte[TableCompactingState._CompactingStateSize];

            _sync.EnterReadLock();
            try
            {
                CompacterRawFileReference.Read(pos, rb, 0, TableCompactingState._CompactingStateSize);
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync.ExitReadLock();
            }

            return TableCompactingState.GetTableCompactingStateFromByteArray(rb);
        }

        /// <summary>
        /// Internal for writing compaction state
        /// </summary>
        /// <param name="ptr"></param>
        /// <param name="tcs"></param>
        private void WriteCompactingState(ulong pos, TableCompactingState tcs)
        {

            _sync.EnterWriteLock();
            try
            {
                byte[] nb = TableCompactingState.GetStateAsByteArray(tcs);

                CompacterRawFileReference.Write(pos, nb, 0, nb.Length);
                CompacterRawFileReference.Flush();

            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync.ExitWriteLock();
            }

        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="started"></param>
        /// <param name="pos"></param>
        private void WriteCompactorHeader(bool started, ulong pos, string originalUserTableName)
        {
            StringBuilder sb = new StringBuilder();

            _sync.EnterWriteLock();
            try
            {

                byte[] nb = TablesCompactorHeader.GetCompactorHeaderAsByteArray(new TablesCompactorHeader()
                {
                    CompactionIsStarted = started,
                    PositionOfCompactionElement = pos,
                    OriginalUserTableName = originalUserTableName
                });

                CompacterRawFileReference.Write(0, nb, 0, nb.Length);
                CompacterRawFileReference.Flush();

            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync.ExitWriteLock();
            }
        }


        /// <summary>
        /// 
        /// </summary>
        /// <param name="pos"></param>
        /// <returns></returns>
        private TablesCompactorHeader ReadCompactorHeader()
        {
            byte[] rb = new byte[TablesCompactorHeader._CompactorHeaderSize];

            _sync.EnterReadLock();
            try
            {
                if (CompacterRawFileReference.Length == 0)
                {
                    return new TablesCompactorHeader()
                    {
                        CompactionIsStarted = false
                    };
                }

                CompacterRawFileReference.Read(0, rb, 0, TablesCompactorHeader._CompactorHeaderSize);
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync.ExitReadLock();
            }

            return TablesCompactorHeader.GetCompactorHeaderFromByteArray(rb);
        }






        /// <summary>
        /// Through this procedure we activate Compactor,
        /// then it works itself, till finished
        /// </summary>
        /// <param name="userTableSchema"></param>
        public void CompactTable(UserTableSchema userTableSchema)
        {
            bool run = false;

            lock (lock_queueTablesToBeCompacted)
            {
                try
                {
                    if (!queueTablesToBeCompacted.Contains(userTableSchema, this._transactionsCoordinator.SchemaCoordinator.StsTablesSchema.UserTableSchemaComparer))
                    {
                        queueTablesToBeCompacted.Enqueue(userTableSchema);
                        //run = true;
                    }

                    //running in any case
                    run = true;
                }
                catch (Exception ex)
                {
                    throw ex;
                }

            }

            if (run)
            {
                lock (lock_CompactorIsActive)
                {
                    if (!CompactorIsActive)
                    {
                        CompactorIsActive = true;
                        Action a = () =>
                        {
                            CompactOneTable();
                        };


                        STSdb.MTL.Utils.Async.AsyncOperations.DoAsync(a);
                        //a.DoAsync();
                    }
                }
            }

        }



        /// <summary>
        /// Only one compacting procedure available per time
        /// </summary>
        bool CompactorIsActive = false;
        object lock_CompactorIsActive = new object();

        /// <summary>
        /// Queue holds table which must be compacted
        /// </summary>
        Queue<UserTableSchema> queueTablesToBeCompacted = new Queue<UserTableSchema>();
        object lock_queueTablesToBeCompacted = new object();

        int sequentialCallsQuantity = 0;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="tableName"></param>
        /// <param name="pos"></param>
        /// <returns></returns>
        private void CompactOneTable()
        {
            TableCompactingResult cr = new TableCompactingResult();

            Action<TableCompactingResult> ReRunCompactor = (tcr) =>
            {
                TablesCompactingResultHasArrived(tcr);
                //CompactOneTable();

                //In case of intensive recursive utilization of the CompactOneTable function we can come to the "object disposed but unreleased from GC#0-2" issue,
                //which can bring to memory leaks, using limited quantity of recursive calls we are absolutely defended from this issue.
                if (sequentialCallsQuantity > 10)
                {
                    sequentialCallsQuantity = 0;
                    lock (lock_CompactorIsActive)
                    {
                        CompactorIsActive = false;
                    }
                }
                else
                {
                    CompactOneTable();
                }
            };


            UserTableSchema userTableSchema = null;
            string tableName = String.Empty;
            ulong pos = 0;


            lock (lock_queueTablesToBeCompacted)
            {

                if (queueTablesToBeCompacted.Count() > 0)
                {
                    userTableSchema = queueTablesToBeCompacted.Dequeue();
                    tableName = userTableSchema.TableName;
                    pos = userTableSchema.CompactedInfoPointer;
                }
                else
                {
                    cr.RealTableName = tableName;
                    lock (lock_CompactorIsActive)
                    {
                        sequentialCallsQuantity = 0;
                        CompactorIsActive = false;
                    }
                    return;
                }
            }


            

            try
            {

                //Getting information about active table
                TableCompactingState tcs = ReadCompactingState(pos);

                if (DateTime.UtcNow.Subtract(tcs.LastCompactionWasMade).TotalMinutes < StsTablesCompactor.COMPACTION_DIFF_BETWEEN_COMPACTS_IN_MINUTES)
                {
                    //Early for compaction
                    ReRunCompactor(cr);
                    return;
                }


                string srcTableName = String.Empty;
                string dstTableName = String.Empty;

                //XTable srcTable = null;
                XTable srcTable = null;
                Locator dstLocator = null;
                Locator srcLocator = null;
                XTable dstTable = null;
                IReadLocator IReadLocator = null;
                IWriteLocator IWriteLocator = null;

                //Can be 0 or 1
                int destRawFileHandlePlace = 0;

                if (tcs.ActiveTable == 1)
                {
                    dstTableName = tableName;
                    srcTableName = "@sys_etn_" + tableName + "/1";
                    destRawFileHandlePlace = 0;
                }
                else
                {
                    srcTableName = tableName;
                    dstTableName = "@sys_etn_" + tableName + "/1";
                    destRawFileHandlePlace = 1;
                }

                srcLocator = new Locator(srcTableName);
                dstLocator = new Locator(dstTableName);



                #region "Check if we need to compact this concrete table"

                //Console.WriteLine(String.Format("CompactCheckStart table {0}; ThreadId: {1}", tableName, System.Threading.Thread.CurrentThread.ManagedThreadId));

                //using original name for the table, cause we want to get its open read locator, there tables are stored with user defined names

                IReadLocator = this._transactionsCoordinator.TablesCoordinator.GetTableAsIReadLocator(tableName);
                if (IReadLocator != null)
                    srcTable = IReadLocator.XTable;
                             

                if (srcTable == null)
                {
                    cr.RealTableName = tableName;
                    ReRunCompactor(cr);
                    return;
                }

                /* Choosing Compacting Startegy */

                cr.RealTableName = tableName;
                //cr.CompactedTableName = exhaledTableName;
                cr.RFS_Before = srcTable.RawFile.Length;
                cr.TS_Before = srcTable.Size;
                cr.RecordsCount = srcTable.Count;

                //Closing locator
                IReadLocator.RemoveReadLocatorsForTheThread(System.Threading.Thread.CurrentThread.ManagedThreadId);

                if (!cr.If_To_Compact_ByExccedingPercentage
                    (TableCompactingResult.CompactingStartegies.Compact_When_WastedToTableSize_ExceedsPercentage, 20)
                    )
                {
                    ReRunCompactor(cr);
                    return;

                }
                /* ------------------------------------------ */
                #endregion



                //Sequential calls grow only in case if the data has to be compacted.
                sequentialCallsQuantity++;

                
                
                using (var tran = this._transactionsCoordinator.GetTransaction())
                {                    
                    tran.RegisterWriteOrSynchroReadTables(tableName);

                    //Console.WriteLine("Compact started" + tableName);


                    //Registering a start of Compacting
                    tcs.CompactingState = TableCompactingState.eState.STOP;
                    WriteCompactingState(pos, tcs);

                    //Registering in the header, that compaction is started
                    WriteCompactorHeader(true, pos, tableName);

                    //Opening table again already inside of lock                   
                    //srcTable = this._transactionsCoordinator.SchemaCoordinator.OpenXTableTechical(srcLocator);

                    //using original name for the table, cause we want to get its open read locator, there tables are stored with user defined names
                    IWriteLocator = this._transactionsCoordinator.TablesCoordinator.GetTableAsIWriteLocator(tableName);
                    if (IWriteLocator != null)
                        srcTable = IWriteLocator.XTable;
                                       


                    /*TEST REGION*/

                    //ulong ttt = tran.Count<long, byte[]>(tableName, false);
                    //Console.WriteLine(String.Format("CompactStart table {1}; Rec: {0}", srcTable.Count,tableName));
                    //if (srcTable.Count != ttt)
                    //{

                    //}
                    /*-*----*/

                    if (srcTable == null)
                    {
                        ReRunCompactor(cr);
                        return;
                    }

                    cr.RealTableName = tableName;
                    cr.RFS_Before = srcTable.RawFile.Length;
                    cr.TS_Before = srcTable.Size;
                    cr.RecordsCount = srcTable.Count;

                    ////CREATING OR OPENING DESTINATION TABLE
                    //tcs.CompactingState = TableCompactingState.eState.DESTINATION_TABLE_IS_CREATING;             //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME)
                    //WriteCompactingState(pos, tcs);

                    dstTable = this._transactionsCoordinator.SchemaCoordinator.CreateOrOpenXTableTechical(dstLocator, srcTable.KeyType, srcTable.RecordType);



                    if (dstTable == null)
                    {
                        ReRunCompactor(cr);
                        return;
                    }

                    dstTable.BlockCapacity = 1;
                    dstTable.CacheCapacity = 1;

                    /*Setting up KeyMap and RecordPersist*/
                    //In STSdb.MTL, persitent layer and complex keys are not stored in database, but supplied as Extra parameters in the Schema
                    //engine.Scheme.SetTableProperties...so they are always available onLine. When we copy table, we try to find out
                    //if we have custom KeyMap or Record Persist elements

                    //ITableProperties itp = this._transactionsCoordinator.TablesCoordinator.GetTableProperties("tableName");
                    ITableProperties itp = this._transactionsCoordinator.TablesCoordinator.GetTableProperties(tableName);

                    if (itp != null && itp.KeyMap != null)
                        dstTable.KeyMap = itp.KeyMap;
                    else
                        dstTable.KeyMap = srcTable.KeyMap;
                                    

                    if (itp != null && itp.RecordPersist != null)
                        dstTable.RecordPersist = itp.RecordPersist;
                    else
                        dstTable.RecordPersist = srcTable.RecordPersist;
                                        

                    if (destRawFileHandlePlace == 0)
                    {
                        tcs.Table0handle = dstTable.RawFile.Handle;
                        tcs.Table1handle = srcTable.RawFile.Handle;
                    }
                    else
                    {
                        tcs.Table1handle = dstTable.RawFile.Handle;
                        tcs.Table0handle = srcTable.RawFile.Handle;
                    }

                    tcs.CompactingState = TableCompactingState.eState.DESTINATION_TABLE_IS_CREATED;           //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME and RAW FILE)
                    WriteCompactingState(pos, tcs);



                    //COPYING THE DATA                    
                    try
                    {
                        //copying old table to the new one
                        foreach (var row in srcTable.Forward())
                            dstTable[row.Key] = row.Record;


                        dstTable.Commit();
                    }
                    catch (Exception ex)
                    {
                        cr.CompacterException = ex;
                        ReRunCompactor(cr);
                        return;
                    }


                    //Configuring compacting result, may be save it later together with DATA_IS_COPIED state
                    cr.RFS_After = dstTable.RawFile.Length;
                    cr.TS_After = dstTable.Size;
                    cr.Compacted = true;

                    tcs.ActiveTable = tcs.ActiveTable * (-1) + 1;       //Switch from 0 to 1 and from 0 to 1, boolean NOT operator.
                    tcs.CompactingState = TableCompactingState.eState.DATA_IS_COPIED;           //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME and RAW FILE), like it's done later in the code.
                    tcs.LastCompactionWasMade = DateTime.UtcNow;                    //Setting up the date of the last compaction
                    WriteCompactingState(pos, tcs);

                    //Deleting table from the TableCoordinator, so newly created READ locators already could refer to the new table
                    DateTime operationCompleted = this._transactionsCoordinator.TablesCoordinator.DeleteTable(tableName);

                    //Waiting till READ locators created before copying of the table finish their job
                    while (this._transactionsCoordinator.ReadLocatorsHandler.IfTableHasOpenReadLocatorsTill(tableName, operationCompleted))
                    {
                        Thread.Sleep(1000);
                    }

                    //if (tableName == "t1")
                    //{
                    //    var si = this._transactionsCoordinator.SchemaCoordinator.GetSchemaInfo();
                    //    Console.WriteLine(String.Format("Compact finished: OrigTN: {0}; SrcTN: {1}; DestTN: {2};", tableName, srcTableName, dstTableName));
                    //    Console.WriteLine(si.ToString());
                    //}

                }//End of using

                

                //DELETING Source Table from the Schema and VFS
                this._transactionsCoordinator.SchemaCoordinator.DeleteXTableTechnical(srcTableName, true);

                tcs.CompactingState = TableCompactingState.eState.SOURCE_TABLE_IS_DELETED;      //IF HERE JUST ChANGE Compacting HEADER ON INCATIVE
                WriteCompactingState(pos, tcs);

                WriteCompactorHeader(false, pos, tableName);


            }
            catch (Exception ex)
            {
                cr.CompacterException = ex;
                ReRunCompactor(cr);
                return;
            }

            ////Giving ability for VFS to finish its job.
            //System.Threading.Thread.Sleep(2000);

            //re-runnig compacter to find another table to be compacted            
            ReRunCompactor(cr);
            return;
        }



        //PREVIOUSLY WORKING COPY
        ///// <summary>
        ///// 
        ///// </summary>
        ///// <param name="tableName"></param>
        ///// <param name="pos"></param>
        ///// <returns></returns>
        //private void CompactOneTable()
        //{
        //    TableCompactingResult cr = new TableCompactingResult();

        //    Action<TableCompactingResult> ReRunCompactor = (tcr) =>
        //    {
        //        TablesCompactingResultHasArrived(tcr);
        //        CompactOneTable();
        //    };


        //    UserTableSchema userTableSchema = null;
        //    string tableName = String.Empty;
        //    ulong pos = 0;


        //    lock (lock_queueTablesToBeCompacted)
        //    {

        //        if (queueTablesToBeCompacted.Count() > 0)
        //        {
        //            userTableSchema = queueTablesToBeCompacted.Dequeue();
        //            tableName = userTableSchema.TableName;
        //            pos = userTableSchema.CompactedInfoPointer;
        //        }
        //        else
        //        {
        //            cr.RealTableName = tableName;
        //            lock (lock_CompactorIsActive)
        //            {
        //                CompactorIsActive = false;
        //            }
        //            return;
        //        }
        //    }

        //    try
        //    {

        //        //Getting information about active table
        //        TableCompactingState tcs = ReadCompactingState(pos);

        //        string srcTableName = String.Empty;
        //        string dstTableName = String.Empty;

        //        XTable srcTable = null;
        //        Locator dstLocator = null;
        //        Locator srcLocator = null;
        //        XTable dstTable = null;

        //        //Can be 0 or 1
        //        int destRawFileHandlePlace = 0;

        //        if (tcs.ActiveTable == 1)
        //        {
        //            dstTableName = tableName;
        //            srcTableName = "@sys_etn_" + tableName + "/1";
        //            destRawFileHandlePlace = 0;
        //        }
        //        else
        //        {
        //            srcTableName = tableName;
        //            dstTableName = "@sys_etn_" + tableName + "/1";
        //            destRawFileHandlePlace = 1;
        //        }

        //        srcLocator = new Locator(srcTableName);
        //        dstLocator = new Locator(dstTableName);



        //        #region "Check if we need to compact this concrete table"

        //        srcTable = this._transactionsCoordinator.SchemaCoordinator.OpenXTableTechical(srcLocator);

        //        if (srcTable == null)
        //        {
        //            cr.RealTableName = tableName;
        //            ReRunCompactor(cr);
        //            return;
        //        }

        //        /* Choosing Compacting Startegy */

        //        cr.RealTableName = tableName;
        //        //cr.CompactedTableName = exhaledTableName;
        //        cr.RFS_Before = srcTable.RawFile.Length;
        //        cr.TS_Before = srcTable.Size;
        //        cr.RecordsCount = srcTable.Count;

        //        if (!cr.If_To_Compact_ByExccedingPercentage
        //            (TableCompactingResult.CompactingStartegies.Compact_When_WastedToTableSize_ExceedsPercentage, 20)
        //            )
        //        {
        //            ReRunCompactor(cr);
        //            return;

        //        }
        //        /* ------------------------------------------ */
        //        #endregion




        //        using (var tran = this._transactionsCoordinator.GetTransaction())
        //        {
        //            tran.RegisterWriteOrSynchroReadTables(srcTableName, dstTableName);


        //            //Registering a start of Compacting
        //            tcs.CompactingState = TableCompactingState.eState.STOP;
        //            WriteCompactingState(pos, tcs);

        //            //Registering in the header, that compaction is started
        //            WriteCompactorHeader(true, pos, tableName);

        //            //Opening table again already inside of lock                   
        //            srcTable = this._transactionsCoordinator.SchemaCoordinator.OpenXTableTechical(srcLocator);

        //            //using original name
        //            //IReadLocator this._transactionsCoordinator.TablesCoordinator.GetTableAsIReadLocator(tableName);


        //            if (srcTable == null)
        //            {
        //                ReRunCompactor(cr);
        //                return;
        //            }

        //            cr.RealTableName = tableName;
        //            cr.RFS_Before = srcTable.RawFile.Length;
        //            cr.TS_Before = srcTable.Size;
        //            cr.RecordsCount = srcTable.Count;

        //            ////CREATING OR OPENING DESTINATION TABLE
        //            //tcs.CompactingState = TableCompactingState.eState.DESTINATION_TABLE_IS_CREATING;             //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME)
        //            //WriteCompactingState(pos, tcs);

        //            dstTable = this._transactionsCoordinator.SchemaCoordinator.CreateOrOpenXTableTechical(dstLocator, srcTable.KeyType, srcTable.RecordType);



        //            if (dstTable == null)
        //            {
        //                ReRunCompactor(cr);
        //                return;
        //            }

        //            dstTable.BlockCapacity = 1;
        //            dstTable.CacheCapacity = 1;

        //            if (srcTable.KeyMap != null)
        //                dstTable.KeyMap = srcTable.KeyMap;

        //            if (srcTable.RecordPersist != null)
        //                dstTable.RecordPersist = srcTable.RecordPersist;

        //            if (destRawFileHandlePlace == 0)
        //            {
        //                tcs.Table0handle = dstTable.RawFile.Handle;
        //                tcs.Table1handle = srcTable.RawFile.Handle;
        //            }
        //            else
        //            {
        //                tcs.Table1handle = dstTable.RawFile.Handle;
        //                tcs.Table0handle = srcTable.RawFile.Handle;
        //            }

        //            tcs.CompactingState = TableCompactingState.eState.DESTINATION_TABLE_IS_CREATED;           //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME and RAW FILE)
        //            WriteCompactingState(pos, tcs);

        //            //COPYING THE DATA                    
        //            try
        //            {
        //                //copying old table to the new one
        //                foreach (var row in srcTable.Forward())
        //                    dstTable[row.Key] = row.Record;

        //                lock (this._transactionsCoordinator._SYNC_COMMIT)
        //                {
        //                    dstTable.Commit();
        //                }
        //            }
        //            catch (Exception ex)
        //            {
        //                cr.CompacterException = ex;
        //                ReRunCompactor(cr);
        //                return;
        //            }


        //            //Configuring compacting result, may be save it later together with DATA_IS_COPIED state
        //            cr.RFS_After = dstTable.RawFile.Length;
        //            cr.TS_After = dstTable.Size;
        //            cr.Compacted = true;

        //            tcs.ActiveTable = tcs.ActiveTable * (-1) + 1;
        //            tcs.CompactingState = TableCompactingState.eState.DATA_IS_COPIED;           //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME and RAW FILE), like it's done later in the code.
        //            WriteCompactingState(pos, tcs);

        //            //Deleting table from the TableCoordinator, so newly created READ locators already could refer to the new table
        //            DateTime operationCompleted = this._transactionsCoordinator.TablesCoordinator.DeleteTable(tableName);

        //            //Waiting till READ locators created before copying of the table finish their job
        //            while (this._transactionsCoordinator.ReadLocatorsHandler.IfTableHasOpenReadLocatorsTill(tableName, operationCompleted))
        //            {
        //                Thread.Sleep(1000);
        //            }


        //        }//End of using

        //        //DELETING Source Table from the Schema and VFS
        //        this._transactionsCoordinator.SchemaCoordinator.DeleteXTableTechnical(srcTableName, true);

        //        tcs.CompactingState = TableCompactingState.eState.SOURCE_TABLE_IS_DELETED;      //IF HERE JUST ChANGE Compacting HEADER ON INCATIVE
        //        WriteCompactingState(pos, tcs);

        //        WriteCompactorHeader(false, pos, tableName);


        //    }
        //    catch (Exception ex)
        //    {
        //        cr.CompacterException = ex;
        //        ReRunCompactor(cr);
        //        return;
        //    }

        //    //re-runnig compacter to find another table to be compacted            
        //    ReRunCompactor(cr);
        //    return;
        //}




        /// <summary>
        /// This function must be called ONLY after the engine start, after initializing of all necessary system tables.
        /// Its purpose is to restore compacting state in case of failure during compacting operation.
        /// </summary>
        public void RestoreCompactingState()
        {
            try
            {
                //Getting compactor header
                TablesCompactorHeader tch = this.ReadCompactorHeader();

                if (!tch.CompactionIsStarted)       //No compaction is started
                    return;

                string tableName = String.Empty;

                //Getting TableCompactingState
                TableCompactingState tcs = this.ReadCompactingState(tch.PositionOfCompactionElement);

                switch (tcs.CompactingState)
                {
                    case TableCompactingState.eState.STOP:
                        //Telling to the header, that everything is alright
                        WriteCompactorHeader(false, 0, tch.OriginalUserTableName);
                        return;
                    case TableCompactingState.eState.DESTINATION_TABLE_IS_CREATED:

                        //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME)

                        if (tcs.ActiveTable == 1)
                        {
                            tableName = tch.OriginalUserTableName;
                        }
                        else
                        {
                            tableName = "@sys_etn_" + tch.OriginalUserTableName + "/1";
                        }

                        this._transactionsCoordinator.SchemaCoordinator.DeleteXTableTechnical(tableName, true);

                        WriteCompactorHeader(false, 0, tch.OriginalUserTableName);
                        return;
                    case TableCompactingState.eState.DATA_IS_COPIED:

                        //IF STOPS HERE (IN-ACTIVE) TABLE MUST BE DROPPED (FROM SCHEMA BY NAME and RAW FILE), like it's done later in the code.

                        if (tcs.ActiveTable == 1)
                        {
                            tableName = tch.OriginalUserTableName;
                        }
                        else
                        {
                            tableName = "@sys_etn_" + tch.OriginalUserTableName + "/1";
                        }

                        this._transactionsCoordinator.SchemaCoordinator.DeleteXTableTechnical(tableName, true);

                        WriteCompactorHeader(false, 0, tch.OriginalUserTableName);

                        return;
                    case TableCompactingState.eState.SOURCE_TABLE_IS_DELETED:
                        //Telling to the header, that everything is alright
                        WriteCompactorHeader(false, 0, tch.OriginalUserTableName);
                        return;

                };
            }
            catch (Exception ex)
            {
                throw ex;
            }

        }


    }
}
